package jp.co.nes.awssupport.control.function;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.amazonaws.services.sqs.model.Message;
import com.amazonaws.services.sqs.model.ReceiveMessageResult;

import jp.co.nes.awssupport.business.bean.CommonContainerBean;
import jp.co.nes.awssupport.business.bean.CommonContainerBean.ProcessingResult;
import jp.co.nes.awssupport.business.bean.ProcessInfoBean;
import jp.co.nes.awssupport.business.logic.ProcessExecutor;
import jp.co.nes.awssupport.business.logic.ProcessExecutor.StartProcessResult;
import jp.co.nes.awssupport.common.base.AbstractRequestHandler;
import jp.co.nes.awssupport.common.base.CommonException;
import jp.co.nes.awssupport.common.utility.AppInfoUtil;
import jp.co.nes.awssupport.common.utility.AwsSqsHelper;
import jp.co.nes.awssupport.common.utility.BeanUtil;

/**
 * 処理管理です(Lambda)。<br>
 * 環境情報に設定されている内容に従い、キューからメッセージを取得し、処理を実行します。
 */
public class ProcessControl extends AbstractRequestHandler<ProcessInfoBean, CommonContainerBean<ProcessInfoBean>> {

    /** ロガー */
    private static Logger logger = LoggerFactory.getLogger(ProcessControl.class);

    /** カテゴリ */
    private static final String CATEGORY = "ProcessControl";

    /** 番号 */
    private int number;

    /** 処理実行 */
    private ProcessExecutor executor;

    /**
     * デフォルトコンストラクタです。
     */
    public ProcessControl() {
    }

    /*
     * (非 Javadoc)
     * @see jp.co.nes.awssupport.common.base.AbstractRequestHandler#handleRequest(java.lang.Object)
     */
    @Override
    protected CommonContainerBean<ProcessInfoBean> handleRequest(ProcessInfoBean input) {
        number = input.getAppInfoNumber();
        String queueName = AppInfoUtil.getValAndEmptyChk(CATEGORY, number, "QueueName");
        int maxNumberOfMessages = AppInfoUtil.getIntValAndEmptyMinChk(CATEGORY, number, "MaxNumberOfMessages", 0);
        String destination = AppInfoUtil.getValAndEmptyChk(CATEGORY, number, "Destination");
        int waitTime = AppInfoUtil.getIntValAndEmptyMinChk(CATEGORY, number, "WaitTime", 0);
        int maxPerSecondStartCount = AppInfoUtil.getIntValAndEmptyMinChk(CATEGORY, number, "MaxPerSecondStartCount", 0);
        int sleepTimeBeforeStart = AppInfoUtil.getIntValAndEmptyMinChk(CATEGORY, number, "SleepTimeBeforeStart", 0);
        if (logger.isDebugEnabled()) {
            logger.debug(CATEGORY + "," + number + ",QueueName=" + queueName);
            logger.debug(CATEGORY + "," + number + ",MaxNumberOfMessages=" + maxNumberOfMessages);
            logger.debug(CATEGORY + "," + number + ",Destination=" + destination);
            logger.debug(CATEGORY + "," + number + ",WaitTime=" + waitTime);
            logger.debug(CATEGORY + "," + number + ",MaxPerSecondStartCount=" + maxPerSecondStartCount);
            logger.debug(CATEGORY + "," + number + ",SleepTimeBeforeStart=" + sleepTimeBeforeStart);
        }

        AwsSqsHelper helper = createAwsSqsHelper();

        executor = createProcessExecutor();
        executor.confirm();
        int processingCount = executor.getProcessingCount();
        int successProcessStartCount = executor.getSuccessProcessStartCount();
        int failureProcessStartCount = executor.getFailureProcessStartCount();
        int unknownCount = executor.getUnknownCount();

        checkUnknownCount(unknownCount);
        checkFailureProcessStartCount(failureProcessStartCount);

        int processableCount = getProcessableCount(processingCount);
        if (isProcessable(processableCount)) {
            // 処理可能な場合

            checkRemainingTime();
            outLoop: while (true) {
                // キューからメッセージを取得し、処理を実行する。
                ReceiveMessageResult receiveMessageResult = helper.receiveMessage(queueName, maxNumberOfMessages);
                List<Message> messageList = receiveMessageResult.getMessages();
                if (messageList.isEmpty()) {
                    if (!isProcessable(processableCount)) {
                        break outLoop;
                    }
                    checkFailureProcessStartCount(failureProcessStartCount);
                    if (!isRemainingTime()) {
                        break outLoop;
                    }
                    continue;
                }

                int perSecoundStartCount = 0;
                long start = System.currentTimeMillis();
                for (Message message : messageList) {
                    perSecoundStartCount++;
                    if (logger.isDebugEnabled()) {
                        logger.debug("perSecoundStartCount=" + perSecoundStartCount);
                    }

                    // 1秒間で処理する最大値以上になった場合、スリープしてから処理を進める。
                    long cur = System.currentTimeMillis();
                    if (cur - start <= 1000) {
                        logger.debug("cur - start <= 1000");
                        if (perSecoundStartCount >= maxPerSecondStartCount) {
                            logger.debug("perSecoundStartCount >= maxPerSecondStartCount");
                            perSecoundStartCount = 1;
                            start = System.currentTimeMillis();
                            sleep(sleepTimeBeforeStart);
                        }
                    } else {
                        logger.debug("cur - start > 1000");
                        perSecoundStartCount = 1;
                        start = System.currentTimeMillis();
                    }

                    // 処理を実行
                    String processName = input.getProcessName();
                    StartProcessResult result = executor.startProcessAsync(processName, destination, message);
                    if (result == StartProcessResult.SUCCESS) {
                        helper.deleteMessage(queueName, message.getReceiptHandle());
                        processableCount--;
                        successProcessStartCount++;
                    } else if (result == StartProcessResult.FAILURE) {
                        failureProcessStartCount++;
                    } else if (result == StartProcessResult.NO_NEED) {
                        // 念のためワーニングログを出力する。
                        logger.warn("startProcessResult=NoNeed(message=" + message + ")");
                    } else if (result == StartProcessResult.NO_NEED_AND_MESSAGE_DELETE) {
                        helper.deleteMessage(queueName, message.getReceiptHandle());
                    } else {
                        // 通常発生しない。
                        throw new CommonException("startProcessResult is wrong");
                    }

                    if (!isProcessable(processableCount)) {
                        break outLoop;
                    }
                    checkFailureProcessStartCount(failureProcessStartCount);
                    if (!isRemainingTime()) {
                        break outLoop;
                    }
                }
            }
        }

        // 返却内容の設定
        ProcessInfoBean info = new ProcessInfoBean();
        BeanUtil.copyProperties(info, input);
        ProcessingResult processingResult = executor.getProcessingResult(successProcessStartCount,
                failureProcessStartCount);
        if (processingResult == ProcessingResult.WAIT) {
            info.setWaitTime(waitTime);
            processingResult = ProcessingResult.WAIT;
        } else {
            info.setWaitTime(0);
        }
        CommonContainerBean<ProcessInfoBean> output = new CommonContainerBean<>();
        output.setInfoBean(info);
        output.setProcessingResult(processingResult);
        return output;
    }

    /*
     * (非 Javadoc)
     * @see jp.co.nes.awssupport.common.base.AbstractRequestHandler#finish(boolean, java.lang.Object, java.lang.Object)
     */
    @Override
    protected void finish(boolean isSuccess, ProcessInfoBean input, CommonContainerBean<ProcessInfoBean> output) {
        try {
            if (executor != null) {
                executor.release();
                executor = null;
            }
        } finally {
            super.finish(isSuccess, input, output);
        }
    }

    /**
     * SQSヘルパーを生成します。
     *
     * @return SQSヘルパー
     */
    protected AwsSqsHelper createAwsSqsHelper() {
        return new AwsSqsHelper();
    }

    /**
     * 処理実行インスタンスを生成します。
     *
     * @return 処理実行インスタンス
     */
    private ProcessExecutor createProcessExecutor() {
        String processExecutorClass = AppInfoUtil.getValAndEmptyChk(CATEGORY, number, "ProcessExecutorClass");
        if (logger.isDebugEnabled()) {
            logger.debug(CATEGORY + "," + number + ",ProcessExecutorClass=" + processExecutorClass);
        }

        try {
            return (ProcessExecutor) Class.forName(processExecutorClass).newInstance();
        } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
            throw new CommonException(e);
        }
    }

    /**
     * 処理実行可能件数を返却します。
     *
     * @param processingCount 呼び出した処理の実行中件数
     * @return 処理実行可能件数
     */
    private int getProcessableCount(int processingCount) {
        int maxProcessableCount = AppInfoUtil.getIntValAndEmptyMinChk(CATEGORY, number, "MaxProcessableCount", 0);
        if (logger.isDebugEnabled()) {
            logger.debug(CATEGORY + "," + number + ",MaxProcessableCount=" + maxProcessableCount);
        }

        return maxProcessableCount - processingCount;
    }

    /**
     * 処理実行が可能かどうかを返却します。
     *
     * @param processableCount 処理実行可能件数
     * @return true: 可, false, 不可
     */
    private boolean isProcessable(int processableCount) {
        if (processableCount <= 0) {
            logger.debug("processableCount == 0");
            return false;
        }
        return true;
    }

    /**
     * 呼び出し処理が失敗した件数が設定以上になったかどうかをチェックします。
     *
     * @param failureCountProcessStartCount 呼び出し処理が失敗した件数
     */
    private void checkFailureProcessStartCount(int failureCountProcessStartCount) {
        int maxFailureCountProcessStartCount = AppInfoUtil.getIntValAndEmptyMinChk(CATEGORY, number,
                "MaxFailureProcessStartCount", 0);
        if (logger.isDebugEnabled()) {
            logger.debug(CATEGORY + "," + number + ",MaxFailureProcessStartCount=" + maxFailureCountProcessStartCount);
        }

        if (failureCountProcessStartCount >= maxFailureCountProcessStartCount) {
            // 失敗した件数が設定値以上になっている場合、例外をスロー
            throw new CommonException("failureCount >= intMaxFailureCount");
        }
    }

    /**
     * 呼び出した処理の状態が不明な件数が設定以上になったかどうかをチェックします。
     *
     * @param unknownCount 呼び出した処理の状態が不明な件数
     */
    private void checkUnknownCount(int unknownCount) {
        int maxUnknownCount = AppInfoUtil.getIntValAndEmptyMinChk(CATEGORY, number, "MaxUnknownCount", 0);
        if (logger.isDebugEnabled()) {
            logger.debug(CATEGORY + "," + number + ",MaxUnknownCount=" + maxUnknownCount);
        }

        if (unknownCount >= maxUnknownCount) {
            // 呼び出した処理の状態が不明な件数が設定値以上になっている場合、例外をスロー
            throw new CommonException("unknownCount >= maxUnknownCount");
        }
    }

    /**
     * 残時間が設定値より大きいかどうかチェックします。
     */
    private void checkRemainingTime() {
        if (!isRemainingTime()) {
            throw new CommonException("remaining time error");
        }
    }

    /**
     * 残時間が設定値より大きいかどうかを返却します。
     *
     * @return true: 大きい, false 大きくない
     */
    private boolean isRemainingTime() {
        long limitOfRemainingTime = AppInfoUtil.getLongValAndEmptyMinChk(CATEGORY, number, "LimitOfRemainingTime", 0L);
        if (logger.isDebugEnabled()) {
            logger.debug(CATEGORY + "," + number + ",LimitOfRemainingTime=" + limitOfRemainingTime);
        }

        int remainingTimeInMillis = context.getRemainingTimeInMillis();
        if (logger.isDebugEnabled()) {
            logger.debug("remainingTimeInMillis=" + remainingTimeInMillis);
        }

        if (remainingTimeInMillis <= limitOfRemainingTime) {
            // 残り処理時間が設定値以下の場合
            logger.debug("remainingTimeInMillis <= longLimitOfRemainingTime");
            return false;
        }
        return true;
    }

    /**
     * スリープ処理です。
     *
     * @param sleepTime スリープ時間
     */
    private void sleep(int sleepTime) {
        try {
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
            logger.warn(e.getMessage());
        }
    }
}
